package org.gxlu.com.jstorm.day1;

import java.util.Map;
import java.util.Random;

import backtype.storm.spout.SpoutOutputCollector;
import backtype.storm.task.TopologyContext;
import backtype.storm.topology.OutputFieldsDeclarer;
import backtype.storm.topology.base.BaseRichSpout;
import backtype.storm.tuple.Fields;
import backtype.storm.tuple.Values;

public class WCSpount  extends BaseRichSpout{

	/**
	 * 
	 */
	private static final long serialVersionUID = 1L;

	SpoutOutputCollector collector = null;
	
	private String[] words = new String[]{"storm hive spark ", "hadoop hbase hive ", 
			"hive hadoop hello world", "flume hello world","hbase storm spark","spark hadoop hive hbase"};
	private Random random = new Random();
	@Override
	public void nextTuple() {
		collector.emit(new Values(words[random.nextInt(words.length)]));
	}

	@Override
	public void open(@SuppressWarnings("rawtypes") Map arg0, TopologyContext arg1, SpoutOutputCollector arg2) {
		 collector =arg2 ;
	}

	@Override
	public void declareOutputFields(OutputFieldsDeclarer declare) {
		declare.declare(new Fields("str"));
	}

}
